# !/usr/bin/env python
# -*- coding: utf-8 -*-
"""
-------------------------------------------------
   File Name：    dingding_notifier.py
   Description:   钉钉通知工具类，适用于企业级测试用例发送通知的需求
-------------------------------------------------
"""

from __future__ import annotations

import requests
import json
import logging
from typing import List, Dict, Any
from pydantic import BaseModel, Field, ValidationError

# 初始化日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class DingDingConfig(BaseModel):
    webhook_url: str = Field(..., description="钉钉群机器人Webhook URL")
    secret: str | None = Field(None, description="钉钉机器人安全设置的密钥，可选")


class DingDingNotifier:
    """
    企业级钉钉通知工具类，支持发送文本和Markdown消息。
    """

    def __init__(self, config: DingDingConfig):
        self.webhook_url = config.webhook_url
        self.secret = config.secret

    def send_text_message(self, content: str, at_mobiles: List[str] = None, is_at_all: bool = False) -> Dict[str, Any]:
        """
        发送文本消息
        :param content: 消息内容
        :param at_mobiles: 需要@的手机号列表
        :param is_at_all: 是否@所有人
        :return: 钉钉接口的响应结果
        """
        message = {
            "msgtype": "text",
            "text": {
                "content": content
            },
            "at": {
                "atMobiles": at_mobiles if at_mobiles else [],
                "isAtAll": is_at_all
            }
        }
        return self._send_message(message)

    def send_markdown_message(self, title: str, content: str, at_mobiles: List[str] = None, is_at_all: bool = False) -> \
            Dict[str, Any]:
        """
        发送Markdown消息
        :param title: 消息标题
        :param content: Markdown格式的消息内容
        :param at_mobiles: 需要@的手机号列表
        :param is_at_all: 是否@所有人
        :return: 钉钉接口的响应结果
        """
        message = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": content
            },
            "at": {
                "atMobiles": at_mobiles if at_mobiles else [],
                "isAtAll": is_at_all
            }
        }
        return self._send_message(message)

    def _send_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
        """
        发送消息到钉钉群
        :param message: 消息内容字典
        :return: 钉钉接口的响应结果
        """
        headers = {
            'Content-Type': 'application/json'
        }
        try:
            response = requests.post(self.webhook_url, headers=headers, data=json.dumps(message))
            response.raise_for_status()
            result = response.json()
            if result.get("errcode") != 0:
                logger.error(f"发送钉钉消息失败: {result}")
            else:
                logger.info(f"钉钉消息发送成功: {result}")
            return result
        except requests.exceptions.RequestException as e:
            logger.error(f"钉钉消息发送失败，错误信息: {e}")
            return {"errcode": -1, "errmsg": str(e)}


# 示例使用
if __name__ == '__main__':
    # 初始化钉钉配置
    try:
        config = DingDingConfig(
            webhook_url="https://oapi.dingtalk.com/robot/send?access_token=YOUR_ACCESS_TOKEN",
            secret=None  # 可选项，如果设置了加签安全，则填写对应的密钥
        )
    except ValidationError as e:
        logger.error(f"配置校验失败: {e}")
        raise

    notifier = DingDingNotifier(config)
    # 发送文本消息
    notifier.send_text_message("测试用例执行成功！", at_mobiles=["1234567890"], is_at_all=False)
    # 发送Markdown消息
    notifier.send_markdown_message("测试报告", "### 测试结果\n- 所有测试用例已通过", at_mobiles=["1234567890"],
                                   is_at_all=False)
